package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TrafficHomeDetailPageViewBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * ClassName: DwsTrafficHomeDetailPageViewWindow
 * Description:
 * 首页、详情页独立访客
 *
 * @Create 2024/4/25 13:44
 */
public class DwsTrafficHomeDetailPageViewWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsTrafficHomeDetailPageViewWindow().start(
                10023,
                4,
                "dws_traffic_home_detail_page_view_window",
                Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaSourceDS) {
        //TODO 1.对流中数据类型进行转换    jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaSourceDS.map(JSON::parseObject);


        //TODO 2.过滤出首页及详情页
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.filter(
                new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObj) throws Exception {

                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                        String pageId = pageJsonObj.getString("page_id");

                        return "home".equals(pageId) || "good_detail".equals(pageId);
                    }
                }
        );

        //filterDS.print();


        //TODO 3.指定Watermark生成策略以及提取事件时间字段
        SingleOutputStreamOperator<JSONObject> withwatermarkDS = filterDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("ts");
                                    }
                                }
                        )
        );


        //TODO 4.按照mid分组
        KeyedStream<JSONObject, String> keyedDS = withwatermarkDS.keyBy(t -> t.getJSONObject("common").getString("mid"));


        //TODO 5.使用状态编程，判断是否为首页、详情页，并封装为实体类对象
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> beanDS = keyedDS.process(
                new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {

                    private ValueState<String> homeLastVisitDateState;

                    private ValueState<String> goodDetailLastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> homeValueStateDescriptor =
                                new ValueStateDescriptor<String>("homeLastVisitDateState", String.class);

                        homeValueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());

                        homeLastVisitDateState = getRuntimeContext().getState(homeValueStateDescriptor);


                        ValueStateDescriptor<String> goodDetailValueStateDescriptor =
                                new ValueStateDescriptor<String>("goodDetailLastVisitDateState", String.class);

                        goodDetailValueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());

                        goodDetailLastVisitDateState = getRuntimeContext().getState(goodDetailValueStateDescriptor);

                    }

                    @Override
                    public void processElement(JSONObject jsonObj,
                                               KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>.Context ctx,
                                               Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                        //获取当日访问日期
                        Long ts = jsonObj.getLong("ts");
                        String curVisitDate = DateFormatUtil.tsToDate(ts);


                        long homeUvCt = 0L;
                        long goodDetailUvCt = 0L;

                        //获取当前访问页面的id
                        String pageId = jsonObj.getJSONObject("page").getString("page_id");

                        if ("home".equals(pageId)) {
                            String homeLastVisitDate = homeLastVisitDateState.value();
                            if (StringUtils.isEmpty(homeLastVisitDate) || !homeLastVisitDate.equals(curVisitDate)) {
                                homeUvCt = 1L;
                                homeLastVisitDateState.update(curVisitDate);
                            }
                        } else {
                            String goodDetailLastVisitDate = goodDetailLastVisitDateState.value();
                            if (StringUtils.isEmpty(goodDetailLastVisitDate) || !goodDetailLastVisitDate.equals(curVisitDate)) {
                                goodDetailUvCt = 1L;
                                goodDetailLastVisitDateState.update(curVisitDate);
                            }
                        }

                        //根据前一个判断，如果当前数据日期和状态日期一样，那么不会将状态改为1，还是0
                        //以下判断为防止下一条相同mid数据重复计算，重复的就不输出对象了
                        if (homeUvCt != 0L || goodDetailUvCt != 0L) {
                            out.collect(
                                    new TrafficHomeDetailPageViewBean(
                                            "",
                                            "",
                                            "",
                                            homeUvCt,
                                            goodDetailUvCt,
                                            ts
                                    )
                            );
                        }


                    }
                }
        );


        //TODO 6.开窗
        AllWindowedStream<TrafficHomeDetailPageViewBean, TimeWindow> windowDS =
                beanDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 7.聚合
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TrafficHomeDetailPageViewBean>() {
                    @Override
                    public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean value1,
                                                                TrafficHomeDetailPageViewBean value2) throws Exception {

                        value1.setHomeUvCt(value1.getHomeUvCt() + value2.getHomeUvCt());
                        value1.setGoodDetailUvCt(value1.getGoodDetailUvCt() + value2.getGoodDetailUvCt());

                        return value1;
                    }
                },
                new ProcessAllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>.Context context,
                                        Iterable<TrafficHomeDetailPageViewBean> elements,
                                        Collector<TrafficHomeDetailPageViewBean> out) throws Exception {

                        String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                        String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                        String curDate = DateFormatUtil.tsToDate(context.window().getStart());

                        TrafficHomeDetailPageViewBean viewBean = elements.iterator().next();

                        viewBean.setStt(stt);
                        viewBean.setEdt(edt);
                        viewBean.setCurDate(curDate);

                        out.collect(viewBean);

                    }
                }
        );


        //TODO 7.写到Doris
        reduceDS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_home_detail_page_view_window"));

    }
}
